package cn.kennylee.codehub.mq.lip.integration.kafka.listener;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.lang.NonNull;

/**
 * 抽象Kafka消息监听器 - 手动提交版本
 * <p>Created on 2025/6/12.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractAckKafkaListener extends AbstractKafkaListener implements AcknowledgingMessageListener<String, String> {

    public AbstractAckKafkaListener(String topic) {
        super(topic);
    }

    public AbstractAckKafkaListener(String topic, String consumerGroup) {
        super(topic, consumerGroup);
    }

    @Override
    public void onMessage(@NonNull ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            boolean success = super.processMessage(record);

            if (success) {
                ack.acknowledge(); // 手动提交
            } else {
                handleFailure(record);
                // 不提交，等待 Kafka 自动重试（需配置 `enable.auto.commit=false`）
            }
        } catch (Exception e) {
            log.error("处理消息异常. Topic: {}", getTopic(), e);
        }
    }
}
